import com.sun.javafx.util.Logging
import org.apache.spark.sql.SparkSession

import util.{Distance, TimeUtil}

/**
  *

  *  出行驻留 第一阶段
  */

object TravelRecognitionModel extends Logging{


 def main(args: Array[String]): Unit = {


   val spark: SparkSession = SparkSession.builder()
     .master("local[*]")
     .appName("jiaotong")
     .getOrCreate()


   //读取清洗后的数据
   spark.read.format("csv")
     .option("delimiter", ",") //分隔符
     .option("qoute", "")
     .option("nullValue", "\\N")
     .load("output3/out/")
     .select("_c0","_c1","_c2","_c3")
     .toDF("time","imsi","longitude","latitude")
     .distinct()  //去重数据
     .createTempView("result")

   //数据准备
 val result_num = spark.sql(
         """
           |
           |select
           |  time,
           |  imsi,
           |  concat(longitude,'-',latitude) as long_lat,
           |  row_number() over (partition by imsi order by time) number
           |from result
         """.stripMargin)

   result_num.createTempView("result_num")

   //TODO 获取出现次数小于4的经纬度点所在的imsi的所有数据

  val result_count =  spark.sql(
     """
       |
       |select
       |  rt.imsi,
       |  rt.long_lat,
       |  count(rt.long_lat) count
       |from (
       |      select
       |         imsi,
       |         concat(longitude,'-',latitude) as long_lat
       |      from result
       |      ) rt
       | group by rt.imsi,rt.long_lat
       | order by rt.imsi,rt.long_lat
       |
     """.stripMargin)
     .where("count<4")
   //验证 正确
   result_count.selectExpr("sum(count)").show()

   /**
     *  出现频率三次以内的点，可能单，可能连续
     */

   //time,imsi,long_lat,number
//  val result_end =  result_num.join(result_count,Seq("imsi","long_lat"),"right")
//                      .selectExpr("time","imsi","long_lat","number").orderBy("imsi","number")
//
//     //TODO  逻辑：判断当前点的前后数是否连续
//
//     val result_list =  result_end.rdd.collect().toList
//     //前后不连续点集合
//     var result_one = scala.collection.mutable.ArrayBuffer[String]()
//
//   //迭代的 前后连续点 数据集
//   var result_more = scala.collection.mutable.ArrayBuffer[String]()
//   //分段后的结果集
//   var result_more_list = List[String]()
//     for(v0 <- 0 until (result_list.length-1)){
//
//       val number_v0 = result_list(v0)(3).toString.toLong
//       //下一个点
//       val number_v1 = number_v0+1
//       //上一个点
//       val number_v2 = number_v0-1
//
//
//       if(v0 ==0){
//
//         val number_v0_plus = result_list(v0+1)(3)
//
//         if(number_v1 != number_v0_plus){
//
//           result_one += result_list(v0)(1)+"-"+result_list(v0)(3)
//
//         }else {
//           result_more += result_list(v0)(0)+"-"+result_list(v0)(1)+"-"+result_list(v0)(2)+"-"+result_list(v0)(3)
//         }
//
//       } else if(v0 ==result_list.length-1){
//
//         val number_v0_min =  result_list(v0-1)(3)
//
//         if ( number_v2 != number_v0_min) {
//
//           result_one += result_list(v0)(1)+"-"+result_list(v0)(3)
//
//
//         }else {
//
//           result_more += result_list(v0)(0)+"-"+result_list(v0)(1)+"-"+result_list(v0)(2)+"-"+result_list(v0)(3)
//
//           result_more_list = result_more_list.::(result_more.toString())
//
//         }
//       }else{
//
//         val number_v0_plus = result_list(v0+1)(3)
//
//         val number_v0_min =  result_list(v0-1)(3)
//
//         if (number_v1 != number_v0_plus && number_v2 != number_v0_min) {
//
//           result_one += result_list(v0)(1)+"-"+result_list(v0)(3)
//
//           result_more.clear()
//
//         }
//         //上一个相同，下一个点不相同
//         else if(number_v1 != number_v0_plus && number_v2 == number_v0_min){
//
//           result_more += result_list(v0)(0)+"-"+result_list(v0)(1)+"-"+result_list(v0)(2)+"-"+result_list(v0)(3)
//
//           result_more_list = result_more_list.::(result_more.toString())
//
//           result_more.clear()
//
//         }
//         else {
//           result_more += result_list(v0)(0)+"-"+result_list(v0)(1)+"-"+result_list(v0)(2)+"-"+result_list(v0)(3)
//         }
//       }
//
//     }
//
//   /**
//     *      判断速度
//     *      将大于 8km/h的点 认定为出行点
//     */
//
//  val result_speed =  scala.collection.mutable.ArrayBuffer[String]()
//
//  val speed_more =  result_more_list.iterator
//
//   while(speed_more.hasNext){
//
//     val number =  speed_more.next()
//     //去掉多余部分 数据清洗
//     val speed_detail = number.toString.replaceAll("ArrayBuffer\\(","").replaceAll("\\)","")
//
//     //split(",") 以逗号分割的list
//     val speed_detail_list = speed_detail.split(",").toList
//
//     val imsi = speed_detail.split(",").toList.head.split("-")(1)
//
//     val nid = speed_detail.split(",").toList.head.split("-")(4)
//
//     //时间  获得时间最小
//     val data1 = speed_detail.split(",").toList.head.split("-")(0).toString
//     //时间 倒序 按照nid排序，获得时间最大
//     val data2 = speed_detail.split(",").toList
//                   .sortBy(x=>x.split("-")(4))
//                   .reverse
//                   .head
//                   .split("-")(0).toString
//
//     // TODO 时间//得去掉边界 略坑   '_20181003203105'
//
//     var distance = 0L
//
//     var  data   = TimeUtil.time(data1.trim,data2.trim)
//
//     for(v<-(0 until(speed_detail_list.length-1))){
//
//     //距离计算 当前点经纬度 20181003111929-460070095040312574-123.4287491-41.82312012-4
//      var lng1 =  speed_detail_list(v).toString.split("-")(2).toDouble
//
//      var lat1 =  speed_detail_list(v).toString.split("-")(3).toDouble
//     //下一点经纬度
//      var lng2 =  speed_detail_list(v+1).toString.split("-")(2).toDouble
//
//      var lat2 =  speed_detail_list(v+1).toString.split("-")(3).toDouble
//
//       distance  += Distance.distance(lng1,lat1,lng2,lat2).toLong/1000
//
//     }
//     // TODO 计算速度！！！
//     var speed = distance/data
//     // TODO 将符合出行的数据存起来
//     if(speed > 8){
//
//       result_speed += speed_detail
//
//     }
//   }
//
//  //数据格式转换
//  val result_travel_list  = result_speed.toString
//                             .replaceAll("ArrayBuffer\\(","").replaceAll("\\)","")
//                               .replaceAll("\\,","\\|").replaceAll("\\-","\\,")
//                             .split("\\|")
//                             .toList
//
//  val rtl_iterator =  result_travel_list.iterator
//
//   /**
//     *计算结果保存，list要先转成rdd
//     */
//   val rt_rdd =  spark.sparkContext.parallelize(result_travel_list)
//
//   ///处理 有空字符的数据
//  val result_rt =  rt_rdd.map(x=>{
//    (x.split(",")(0).toString.trim,x.split(",")(1),x.split(",")(2),
//      x.split(",")(3),x.split(",")(4))
//   })
//
//   result_rt.map(x=>x.toString().replaceAll("\\(","").replaceAll("\\)",""))
//     .coalesce(1).saveAsTextFile("hdfs://"+args(0)+":9000"+result_path)

 }

}
